package com.atguigu.flink0922.chapter06;

import com.atguigu.flink0922.bean.MarketingUserBehavior;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/3 16:00
 */
public class Flink03_Project_AppStats_With_Channel {
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        // 统计每个渠道 所有用户的每种行为的指标
        env
            .addSource(new MarketSource())
//            .map(mub -> Tuple2.of(mub.getChannel() + "_" + mub.getBehavior(), 1L)).returns(TUPLE(STRING, LONG))
            .map(mub -> Tuple2.of(Tuple2.of(mub.getChannel(), mub.getBehavior()), 1L)).returns(TUPLE(TUPLE(STRING, STRING), LONG))
            // ((channel, behavior) , 1L)   -> (channel_behavior, 1L)
            .keyBy(new KeySelector<Tuple2<Tuple2<String, String>, Long>, Tuple2<String, String>>() {
                @Override
                public Tuple2<String, String> getKey(Tuple2<Tuple2<String, String>, Long> value) throws Exception {
                    return value.f0;
                }
            })
            .sum(1)
            .print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static class MarketSource implements SourceFunction<MarketingUserBehavior> {
        boolean isCancel = false;
        
        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            final Random random = new Random();
            final String[] behaviors = {"download", "install", "uninstall", "update"};
            final String[] channels = {"huawei", "xiaomi", "apple", "vivo", "oppo", "wandoujian", "yingyongbao"};
            while (!isCancel) {
                Long userId = (long) random.nextInt(1000);
                String behavior = behaviors[random.nextInt(behaviors.length)];
                String channel = channels[random.nextInt(channels.length)];
                Long timestamp = System.currentTimeMillis();
    
                ctx.collect(new MarketingUserBehavior(userId, behavior, channel, timestamp));
                Thread.sleep(random.nextInt(500) + 500);
            }
        }
        
        @Override
        public void cancel() {
            isCancel = true;
        }
    }
}
